package com.etlmaster.executor.processors.Table;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.etlmaster.executor.utils.DpExecutorUtils;

/**
 * @author Wugf
 * @date 2016-03-08
 */
public class RationalDbProcessor extends CreateTableProcessor {
	
	public void buildDDl() {
		meta.setTableName(replaceTableName(meta.getTableName(), meta.getDsType()));
		StringBuilder sb = new StringBuilder();
		if (meta.getDsType().equalsIgnoreCase("mysql")) {
			sb.append("create table IF NOT EXISTS ").append(meta.getTableName())
					.append("(");
		} else if(meta.getDsType().equals("db2")&&cfgObj.get("schemaName")!=null&&cfgObj.get("schemaName").toString().length()>0){
			sb.append("create table ").append(cfgObj.get("schemaName")+".").append(meta.getTableName()).append("(");
		}else{
			sb.append("create table ").append(meta.getTableName()).append("(");
		}

		StringBuilder db2Parkey = new StringBuilder();
		StringBuilder praCols=new StringBuilder();
		praCols.append(",primary key(");
		meta.setColumns((List<Map<String, Object>>) cfgObj.get("columns"));
		for (Map<String, Object> aColumn : meta.getColumns()) {
			String colName = (String) aColumn.get("colname");
			String dataType = (String) aColumn.get("datatype");
			String praKey = (String) aColumn.get("key_seq");// 主键;
			if (aColumn.get("length") != null
					&& !aColumn.get("length").toString().equals("")
					&& !aColumn.get("length").toString().equals("0")) {
				dataType += "(" + aColumn.get("length").toString();
				if(aColumn.get("precision_val") != null
						&& !aColumn.get("precision_val").toString().equals("")
						&& !aColumn.get("precision_val").toString().equals("0"))
				{
					dataType+=","+aColumn.get("precision_val").toString();
				}
				dataType+= ") ";
			}
			if(praKey.equals("1")){//是主键
				dataType += " not null ";
				praCols.append(colName+",");
			}
			String partySeq = "";
			if (aColumn.get("party_seq") != null) {
				partySeq = aColumn.get("party_seq").toString();
			}

			if (!"".equals(partySeq)) {// 处理分区键
				db2Parkey.append(db2Parkey.length() > 0 ? "," : "")
						.append(colName).append(" ");
			}
			sb.append(colName).append(" ").append(dataType).append(",");

		}
		praCols.deleteCharAt(praCols.length()-1);
		
		sb.deleteCharAt(sb.length() - 1);
		if(praCols.length()>13){
			praCols.append(")");
			sb.append(praCols);
		}
		sb.append(")");
		if (db2Parkey.length() > 0) {
			meta.setDb2DistributionKey(db2Parkey.toString());
		}
		sb.append(genSuffix(meta.getDsType()));

		if (!meta.getDsType().equals("hive")) {
			String tabSpace = meta
					.variableSubstitution(meta.getTbsSpace() == null ? ""
							: meta.getTbsSpace().trim());// 转换变量空间
			if (tabSpace.length() > 0) {
				sb.append(" in " + tabSpace);
			}
			String indexSpace = meta
					.variableSubstitution(meta.getIndexSpace() == null ? ""
							: meta.getIndexSpace().trim());// 转换变量索引空间
			if (indexSpace.length() > 0) {
				sb.append(" INDEX IN " + indexSpace
						+ " NOT LOGGED INITIALLY COMPRESS YES");
			}
		}
		ddl = sb.toString();

	}

	static Map<String, String> DDL_SUFFIX = new HashMap<String, String>();

	static {
		DDL_SUFFIX
				.put("db2",
						" <#if db2DistributionKey?? > distribute by ( ${db2DistributionKey}) </#if> ");
		DDL_SUFFIX
				.put("postgresql",
						" <#if db2DistributionKey?? > distribute by ( ${db2DistributionKey}) </#if> ");
	}

	private String genSuffix(String dsType) {
		String result = "";
		if (DDL_SUFFIX.containsKey(dsType)) {
			result = DpExecutorUtils.variableSubstitution(
					DDL_SUFFIX.get(dsType), meta);
		}
		return result;
	}

	private String replaceTableName(String tableName, String dsType) {
		String taskId = (String) meta.getTaskid();
		tableName = meta.variableSubstitution(tableName);
		return tableName.replaceAll("YYYYMMDD", taskId).replaceAll("YYYYMM",
				taskId);
	}

}
